package com.pxene.dmp.reducer;

import java.io.IOException;
import java.util.Date;
import java.util.Random;
import java.util.TreeMap;
import java.util.UUID;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class SortFavouriteSiteReducer extends Reducer<Text, Text, LongWritable, Text> {
	private LongWritable resultKey = new LongWritable();
	private Text resultVal = new Text();
	private TreeMap<Long, String> treeMap = new TreeMap<Long, String>();
	private static final int TOP = 10;
	
	
	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		String[] columns = key.toString().split("\\|");
		if (columns.length != 3) {
			return;
		}
		String imsi = columns[0];
		String province = columns[1];
		String dataType = columns[2];
		
		for (Text value : values) {
			String[] array = value.toString().split("\\|");
			if (array.length != 2) {
				continue;
			}
			String domain = array[0];
			long sum = Long.parseLong(array[1]);
			treeMap.put(sum, domain);
			if (treeMap.size() > TOP) {
				treeMap.remove(treeMap.firstKey());
			}
		}

		StringBuilder tags = new StringBuilder();
		for (String domain : treeMap.values()) {
			tags.append(domain).append(";");
		}
		
		int length = tags.length();
		if (length > 1) {
			resultKey.set(new Date().getTime());
			resultVal.set(imsi + "|" + tags.substring(0, length-1) + "|" + province + "|" + dataType);
			context.write(resultKey, resultVal);
		}
	}
	
}
